package com.example.klinemqflink.fun;

import com.example.klinemqflink.model.Kline;
import com.example.klinemqflink.model.MarketData;
import lombok.Data;
import org.apache.flink.api.common.functions.AggregateFunction;
@Data
public class SessionKLineAggregateFunction implements AggregateFunction<MarketData, Kline, Kline> {
    private String frequency;

    public SessionKLineAggregateFunction(String frequency) {
        this.frequency = frequency;
    }

    @Override
    public Kline createAccumulator() {
        return new Kline();
    }

    @Override
    public Kline add(MarketData value, Kline accumulator) {
        if (accumulator.getStartTime() == 0) {
            accumulator.setSymbol(value.getSymbol());
            accumulator.setStartTime(value.getTimestamp());
            accumulator.setOpen(value.getPrice());
            accumulator.setHigh(value.getPrice());
            accumulator.setLow(value.getPrice());
        }
        accumulator.setHigh(Math.max(accumulator.getHigh(), value.getPrice()));
        accumulator.setLow(Math.min(accumulator.getLow(), value.getPrice()));
        accumulator.setClose(value.getPrice());
        return accumulator;
    }

    @Override
    public Kline getResult(Kline accumulator) {
        accumulator.setEndTime(accumulator.getStartTime() + (accumulator.getEndTime() == 0? 0 : accumulator.getEndTime() - accumulator.getStartTime()));
        return accumulator;
    }

    @Override
    public Kline merge(Kline a, Kline b) {
        a.setHigh(Math.max(a.getHigh(), b.getHigh()));
        a.setLow(Math.min(a.getLow(), b.getLow()));
        a.setEndTime(Math.max(a.getEndTime(), b.getEndTime()));
        a.setClose(b.getClose());
        return a;
    }
}
